package com.roy.rocket.config;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.StringMessageConverter;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author ：楼兰
 * @date ：Created in 2020/11/5
 * @description:
 **/

//@RocketMQTransactionListener(txProducerGroup = "springBootGroup2")
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class MyTransactionImpl implements RocketMQLocalTransactionListener {

    private ConcurrentHashMap<Object, String> localTrans = new ConcurrentHashMap<>();
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        Object id = msg.getHeaders().get("id");
        String destination = arg.toString();
        localTrans.put(id,destination);
        org.apache.rocketmq.common.message.Message message = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(),"UTF-8",destination, msg);
        String tags = message.getTags();
        if(StringUtils.contains(tags,"TagA")){
            return RocketMQLocalTransactionState.COMMIT;
        }else if(StringUtils.contains(tags,"TagB")){
            return RocketMQLocalTransactionState.ROLLBACK;
        }else{
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        //SpringBoot的这个版本，消息的ID在重新投递时都变了。不能用原生的那种方式。
//        Object id = msg.getHeaders().get("id");
//        String destination = localTrans.get(id);
//        String[] tempArr = destination.split(":", 2);
//        String topic = tempArr[0];
//        String tags = "";
//        if (tempArr.length > 1) {
//            tags = tempArr[1];
//        }
//        if(StringUtils.contains(tags,"TagC")){
//            return RocketMQLocalTransactionState.COMMIT;
//        }else if(StringUtils.contains(tags,"TagD")){
//            return RocketMQLocalTransactionState.ROLLBACK;
//        }else{
//            return RocketMQLocalTransactionState.UNKNOWN;
//        }
        return RocketMQLocalTransactionState.COMMIT;
    }
}
